package com.link.serve.kafka.consumer;

import com.link.serve.constants.LinkConstant;
import com.link.serve.constants.RedisOperatorConstant;
import com.link.serve.kafka.proto.KafkaMessageProto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

@Slf4j
@Component
public class KafkaMessageConsumer {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 订阅多个 topic（支持通配符匹配多个队列）
    @KafkaListener(
            groupId = "webGroup",
            containerFactory = "kafkaBatchListenerContainerFactory",
            topics ={"user-topic"}
    )
    public void listen(List<KafkaMessageProto.KafkaMessage> messages, Acknowledgment acknowledgment) {
        try {
            if (messages == null || messages.isEmpty()) {acknowledgment.acknowledge();return;}
            batchProcessRedis(messages);// Redis 批量处理
            acknowledgment.acknowledge();// 确认 Kafka 批量消费完成
            log.info("msg_batch✅_num:{}", messages.size());
        } catch (Exception e) {
            log.error("msg_batch❌,skip_next_record: {}", e.getMessage(), e);
        }
    }

    //使用 Redis Pipeline 批量处理消息，优化 Redis 性能
    private void batchProcessRedis(List<KafkaMessageProto.KafkaMessage> messages) {
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (KafkaMessageProto.KafkaMessage message : messages) {
                switch (message.getOperationType()) {
                    // case RedisOperatorConstant.LOGOUT:用户登录
                    case RedisOperatorConstant.LOGIN:
                        connection.stringCommands().setEx(
                                Objects.requireNonNull(redisTemplate.getStringSerializer().serialize(message.getKey())),
                                LinkConstant.REDIS_ONLINE_TIMEOUT_SECOND,
                                Objects.requireNonNull(redisTemplate.getStringSerializer().serialize(String.valueOf(message.getServerId())))
                        );break;
                    case RedisOperatorConstant.HEARTBEAT:
                        connection.keyCommands().expire(
                                Objects.requireNonNull(redisTemplate.getStringSerializer().serialize(message.getKey())),
                                LinkConstant.REDIS_ONLINE_TIMEOUT_SECOND
                        );break;
                    case RedisOperatorConstant.LOGOUT: break;
                    default: break;
                }
            }
            return null;
        });
        log.info("redis_execute_pipeline✅_msgNum:{}", messages.size());
    }
}